package br.pucrs.lognplay.hadoop;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.log4j.Logger;

public class HDFSTools {

	private static final Logger logger = Logger.getLogger(HDFSTools.class);

	public static final String CORE_SITE = "/usr/local/hadoop/conf/core-site.xml";

	public static final String MAPRED_SITE = "/usr/local/hadoop/conf/mapred-site.xml";

	public static final String HDFS_SITE = "/usr/local/hadoop/conf/hdfs-site.xml";

	private static Configuration configuration = null;

	private static FileSystem fileSystem = null;

	public static Configuration getConfiguration() {
		if (configuration == null) {
			configuration = new Configuration();
			configuration.addResource(new Path(CORE_SITE));
			configuration.addResource(new Path(HDFS_SITE));
			configuration.addResource(new Path(MAPRED_SITE));
		}
		return configuration;
	}

	public static void setConfiguration(Configuration configuration) {
		HDFSTools.configuration = configuration;
	}

	public static FileSystem getFileSystem() throws IOException {
		if (fileSystem == null) {
			fileSystem = FileSystem.get(getConfiguration());
		}
		return fileSystem;
	}

	public static void rm(String file) throws IOException {
		logger.trace("removing file: " + file);
		FileSystem fileSystem = getFileSystem();
		Path path = new Path(file);
		if (!fileSystem.exists(path)) {
			logger.warn("File " + file + " does not exists");
			return;
		}
		fileSystem.delete(new Path(file), true);
	}

	public static void mkdir(String dir) throws IOException {
		logger.trace("creating directory: " + dir);
		FileSystem fileSystem = getFileSystem();
		Path path = new Path(dir);
		if (fileSystem.exists(path)) {
			logger.warn("Dir " + dir + " already exists");
			return;
		}
		fileSystem.mkdirs(path);
	}

	/**
	 * TODO - Remove the methods below
	 * 
	 */

	public static void addFile(String source, String dest) throws IOException {
		FileSystem fileSystem = getFileSystem();

		// Get the filename out of the file path
		String filename = source.substring(source.lastIndexOf('/') + 1, source.length());

		// Create the destination path including the filename.
		if (dest.charAt(dest.length() - 1) != '/') {
			dest = dest + "/" + filename;
		} else {
			dest = dest + filename;
		}

		Path path = new Path(dest);
		if (fileSystem.exists(path)) {

			System.out.println("File " + dest + " already exists");
			return;
		}

		FSDataOutputStream out = fileSystem.create(path);
		InputStream in = new BufferedInputStream(new FileInputStream(new File(source)));

		byte[] b = new byte[1024];
		int numBytes = 0;
		while ((numBytes = in.read(b)) > 0) {
			out.write(b, 0, numBytes);
		}

		in.close();
		out.close();
	}

	public static void readFile(String file) throws IOException {
		FileSystem fileSystem = getFileSystem();

		Path path = new Path(file);
		if (!fileSystem.exists(path)) {
			System.out.println("File " + file + " does not exists");
			return;
		}

		FSDataInputStream in = fileSystem.open(path);

		String filename = file.substring(file.lastIndexOf('/') + 1, file.length());

		OutputStream out = new BufferedOutputStream(new FileOutputStream(new File(filename)));

		byte[] b = new byte[1024];
		int numBytes = 0;
		while ((numBytes = in.read(b)) > 0) {
			out.write(b, 0, numBytes);
		}

		in.close();
		out.close();
	}

}
